package com.anmi.transfer.table.base;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.anmi.transfer.canal.BaseCanalClient;
import lombok.extern.slf4j.Slf4j;

import java.util.List;

@Slf4j
public abstract class AbstractSyncTable extends BaseCanalClient implements SyncTable {

  /** 所有AbstractSyncTable的子类必须覆盖该属性 */
  public String TABLE_NAME = "";

  @Override
  public void handleData(List<CanalEntry.Entry> entrys) {
    for (CanalEntry.Entry entry : entrys) {
      if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
        CanalEntry.RowChange rowChage;
        try {
          rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
        } catch (Exception e) {
          throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
        }
        // 获取表名
        String tableName = entry.getHeader().getTableName();
        CanalEntry.EventType eventType = rowChage.getEventType();
        if ("".equals(TABLE_NAME)) {
          throw new RuntimeException("table name is empty, please specify it !");
        }
        if (!tableName.equalsIgnoreCase(TABLE_NAME)) {
          break;
        }
        // 打印日志
        logger.info(
            row_format,
            new Object[] {
              entry.getHeader().getLogfileName(),
              String.valueOf(entry.getHeader().getLogfileOffset()),
              entry.getHeader().getSchemaName(),
              entry.getHeader().getTableName(),
              eventType,
              String.valueOf(entry.getHeader().getExecuteTime()),
              entry.getHeader().getGtid()
            });
        // rowChage可以有多个数据行rowData，而一个rowData有多个列名
        for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
          if (eventType == CanalEntry.EventType.INSERT) {
            // MySQL的新增操作，canal可提供修改后的数据
            handleTableNameAndColumns(tableName, rowData.getAfterColumnsList(), eventType);
          } else if (eventType == CanalEntry.EventType.UPDATE) {
            // MySQL的修改操作，canal可提供修改后的数据
            handleTableNameAndColumns(tableName, rowData.getAfterColumnsList(), eventType);
          }
        }
      }
    }
  }

  private void handleTableNameAndColumns(
      String tableName, List<CanalEntry.Column> columns, CanalEntry.EventType eventType) {
    for (CanalEntry.Column column : columns) {
      if (column.getName().equalsIgnoreCase("id") && TABLE_NAME.equalsIgnoreCase(tableName)) {
        if (eventType == CanalEntry.EventType.INSERT) {
          insert(column);
        } else if (eventType == CanalEntry.EventType.UPDATE) {
          update(column);
        }
      }
    }
  }

  public abstract void insert(CanalEntry.Column column);

  public abstract void update(CanalEntry.Column column);
}
